using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Reactive.Threading.Tasks;
using Microsoft.Extensions.Logging;
using ReactiveChannelMessaging.Core;

namespace ReactiveChannelMessaging.Processors;

/// <summary>
/// 消息处理器基类
/// </summary>
public abstract class MessageProcessor<T> : IAsyncDisposable where T : ReactiveChannelMessaging.Core.Message
{
    protected readonly ILogger Logger;
    private readonly Subject<T> _processedMessages;
    private readonly List<IDisposable> _subscriptions;

    protected MessageProcessor(ILogger logger)
    {
        Logger = logger;
        _processedMessages = new Subject<T>();
        _subscriptions = new List<IDisposable>();
    }

    /// <summary>
    /// 已处理的消息流
    /// </summary>
    public IObservable<T> ProcessedMessages => _processedMessages.AsObservable();

    /// <summary>
    /// 订阅消息流并开始处理
    /// </summary>
    public void Subscribe(IObservable<T> messageStream, int concurrency = 1)
    {
        var subscription = messageStream
            .Select(message => Observable.FromAsync(() => ProcessMessageAsync(message)))
            .Merge(concurrency)
            .Subscribe(
                processedMessage => _processedMessages.OnNext(processedMessage),
                error => Logger.LogError(error, "Error in message processor"),
                () => Logger.LogInformation("Message processor completed")
            );

        _subscriptions.Add(subscription);
    }

    /// <summary>
    /// 处理单个消息
    /// </summary>
    protected abstract Task<T> ProcessMessageAsync(T message);

    public virtual async ValueTask DisposeAsync()
    {
        foreach (var subscription in _subscriptions)
        {
            subscription.Dispose();
        }
        _subscriptions.Clear();
        
        _processedMessages.Dispose();
        await Task.CompletedTask;
    }
}

/// <summary>
/// 数据消息处理器
/// </summary>
public class DataMessageProcessor : MessageProcessor<DataMessage>
{
    public DataMessageProcessor(ILogger<DataMessageProcessor> logger) : base(logger)
    {
    }

    protected override async Task<DataMessage> ProcessMessageAsync(DataMessage message)
    {
        Logger.LogInformation("Processing data message: {MessageId}, Content: {Content}", 
            message.Id, message.Content);

        // 模拟数据处理延迟
        await Task.Delay(Random.Shared.Next(10, 100));

        // 数据转换示例
        var processedContent = $"Processed: {message.Content.ToUpperInvariant()}";
        
        var processedMessage = message with 
        { 
            Content = processedContent,
            Timestamp = DateTime.UtcNow
        };

        Logger.LogDebug("Data message processed: {MessageId}", message.Id);
        return processedMessage;
    }
}

/// <summary>
/// 事件消息处理器
/// </summary>
public class EventMessageProcessor : MessageProcessor<EventMessage>
{
    private readonly Subject<string> _eventTypeSubject;

    public EventMessageProcessor(ILogger<EventMessageProcessor> logger) : base(logger)
    {
        _eventTypeSubject = new Subject<string>();
    }

    /// <summary>
    /// 事件类型流
    /// </summary>
    public IObservable<string> EventTypes => _eventTypeSubject.AsObservable();

    protected override async Task<EventMessage> ProcessMessageAsync(EventMessage message)
    {
        Logger.LogInformation("Processing event message: {MessageId}, EventType: {EventType}", 
            message.Id, message.EventType);

        // 发布事件类型到流中
        _eventTypeSubject.OnNext(message.EventType);

        // 模拟事件处理
        await Task.Delay(Random.Shared.Next(5, 50));

        Logger.LogDebug("Event message processed: {MessageId}", message.Id);
        return message;
    }

    public override async ValueTask DisposeAsync()
    {
        _eventTypeSubject.Dispose();
        await base.DisposeAsync();
    }
}

/// <summary>
/// 命令消息处理器
/// </summary>
public class CommandMessageProcessor : MessageProcessor<CommandMessage>
{
    private readonly ReactiveMessageBus _messageBus;

    public CommandMessageProcessor(
        ILogger<CommandMessageProcessor> logger, 
        ReactiveMessageBus messageBus) : base(logger)
    {
        _messageBus = messageBus;
    }

    protected override async Task<CommandMessage> ProcessMessageAsync(CommandMessage message)
    {
        Logger.LogInformation("Processing command message: {MessageId}, Command: {Command}", 
            message.Id, message.Command);

        try
        {
            // 模拟命令执行
            var result = await ExecuteCommandAsync(message);
            
            // 发送响应消息
            var response = new ResponseMessage(
                Guid.NewGuid(),
                DateTime.UtcNow,
                "CommandProcessor",
                message.Id,
                true,
                result
            );

            await _messageBus.PublishAsync(response);
            Logger.LogDebug("Command executed successfully: {MessageId}", message.Id);
        }
        catch (Exception ex)
        {
            Logger.LogError(ex, "Command execution failed: {MessageId}", message.Id);
            
            // 发送错误响应
            var errorResponse = new ResponseMessage(
                Guid.NewGuid(),
                DateTime.UtcNow,
                "CommandProcessor",
                message.Id,
                false,
                null,
                ex.Message
            );

            await _messageBus.PublishAsync(errorResponse);
        }

        return message;
    }

    private async Task<object> ExecuteCommandAsync(CommandMessage command)
    {
        // 模拟不同命令的执行
        return command.Command.ToLowerInvariant() switch
        {
            "calculate" => await CalculateAsync(command.Parameters),
            "fetch_data" => await FetchDataAsync(command.Parameters),
            "transform" => await TransformAsync(command.Parameters),
            _ => throw new NotSupportedException($"Command '{command.Command}' is not supported")
        };
    }

    private async Task<object> CalculateAsync(Dictionary<string, object>? parameters)
    {
        await Task.Delay(50); // 模拟计算延迟
        
        if (parameters?.TryGetValue("operation", out var operation) == true &&
            parameters.TryGetValue("x", out var x) &&
            parameters.TryGetValue("y", out var y))
        {
            return operation.ToString() switch
            {
                "add" => Convert.ToDouble(x) + Convert.ToDouble(y),
                "multiply" => Convert.ToDouble(x) * Convert.ToDouble(y),
                _ => throw new ArgumentException($"Unknown operation: {operation}")
            };
        }
        
        throw new ArgumentException("Invalid parameters for calculate command");
    }

    private async Task<object> FetchDataAsync(Dictionary<string, object>? parameters)
    {
        await Task.Delay(100); // 模拟数据获取延迟
        
        var id = parameters?.TryGetValue("id", out var value) == true ? value.ToString() : "default";
        return new { Id = id, Data = $"Sample data for {id}", Timestamp = DateTime.UtcNow };
    }

    private async Task<object> TransformAsync(Dictionary<string, object>? parameters)
    {
        await Task.Delay(30); // 模拟转换延迟
        
        var input = parameters?.TryGetValue("input", out var value) == true 
            ? value.ToString() 
            : throw new ArgumentException("Input parameter is required");
            
        return new { Original = input, Transformed = input.ToUpperInvariant().Reverse().ToArray() };
    }
}